#include "threadpool.h"
#include <pthread.h>

int threadpoolInit(threadpool_t * tp, int num)
{
    assert(tp != NULL);
    tp->pthreads = (pthread_t*)calloc(num, sizeof(pthread_t));
    tp->threadNumber = num;

    queueInit(&tp->queue);
    return 0;
}

int threadpoolDestroy(threadpool_t * tp)
{
    assert(tp != NULL);
    free(tp->pthreads);
    queueDestroy(&tp->queue);
    return 0;
}


//每一个子线程的入口函数
void * threadFunc(void* arg)
{
    threadpool_t * tp = (threadpool_t*)(arg);
    printf("sub thread start running.\n");
    while(1) {
        int peerfd = taskDequeue(&tp->queue);
        if(peerfd > 0) {
            //执行文件的发送操作
            transferFile(peerfd);
        } else {
            break;//子线程就会退出
        }
    }
    printf("sub thread %ld exit.\n", pthread_self());
    return NULL;
}


//开启线程池,让每一个子线程运行起来
int threadpoolStart(threadpool_t * tp)
{
    assert(tp != NULL);
    for(int i = 0; i < tp->threadNumber; ++i) {
        int ret = pthread_create(&tp->pthreads[i], NULL, threadFunc, tp);
        THREAD_ERROR_CHECK(ret, "pthread_create");
    }
    return 0;
}

int threadpoolStop(threadpool_t * tp)
{
    //当任务队列中不为空的情况下，继续执行任务
    while(!queueEmptyWithlock(&tp->queue)) {
        sleep(1);
    }

    //先唤醒所有子线程
    wakeup(&tp->queue);
    printf("wakeup.\n");
    for(int i = 0; i < tp->threadNumber; ++i) {
        pthread_join(tp->pthreads[i], NULL);
    }
    return 0;
}
